センサーデータを分析するデータ基盤候補として、Data Firehose + S3 Tables + Athenaを試してみた

export AWS_PROFILE={AWSプロファイル名}
export AWS_REGION=us-east-1 

spark-shell \
--packages org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.6.1,software.amazon.s3tables:s3-tables-catalog-for-iceberg-runtime:0.1.3,software.amazon.awssdk:s3tables:2.29.26,software.amazon.awssdk:s3:2.29.26,software.amazon.awssdk:sts:2.29.26,software.amazon.awssdk:kms:2.29.26,software.amazon.awssdk:dynamodb:2.29.26,software.amazon.awssdk:kms:2.29.26,software.amazon.awssdk:glue:2.29.26 \
--conf spark.sql.catalog.s3tablesbucket=org.apache.iceberg.spark.SparkCatalog \
--conf spark.sql.catalog.s3tablesbucket.catalog-impl=software.amazon.s3tables.iceberg.S3TablesCatalog \
--conf spark.sql.catalog.s3tablesbucket.warehouse={S3 TablesバケットのARN} \
--conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions

spark.sql("create namespace if not exists s3tablesbucket.sample")

spark.sql("create table if not exists s3tablesbucket.sample.sensor_readings ( sensor_id string, temperature float, humidity float, timestamp timestamp) using iceberg ")

spark.sql("select * from s3tablesbucket.sample.sensor_readings limit 5").show()

+---------+-----------+--------+---------+
|sensor_id|temperature|humidity|timestamp|
+---------+-----------+--------+---------+
+---------+-----------+--------+---------+

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "sts:AssumeRole"
            ],
            "Principal": {
                "Service": [
                    "firehose.amazonaws.com"
                ]
            }
        }
    ]
}

{
	"Version": "2012-10-17",
	"Statement": [
		{
			"Effect": "Allow",
			"Action": [
				"glue:GetTable",
				"glue:GetDatabase",
				"glue:UpdateTable"
			],
			"Resource": [
				"*"
			]
		},
		{
			"Effect": "Allow",
			"Action": [
				"s3:AbortMultipartUpload",
				"s3:GetBucketLocation",
				"s3:GetObject",
				"s3:ListBucket",
				"s3:ListBucketMultipartUploads",
				"s3:PutObject"
			],
			"Resource": [
				"arn:aws:s3:::{Firehoseストリームでエラーになったとき用のS3バケット名}",
				"arn:aws:s3:::{Firehoseストリームでエラーになったとき用のS3バケット名}/*"
			]
		},
		{
			"Effect": "Allow",
			"Action": [
				"lakeformation:GetDataAccess"
			],
			"Resource": "*"
		},
		{
			"Effect": "Allow",
			"Action": [
				"logs:PutLogEvents"
			],
			"Resource": [
				"arn:aws:logs:us-east-1:{アカウント番号}:log-group:*"
			]
		}
	]
}

aws glue create-database --database-input '{
  "Name": "resource_link_to_example",
  "TargetDatabase": {
    "CatalogId": "{アカウント番号}:s3tablescatalog/analysis-test",
    "DatabaseName": "example"
  }
}'

#!/bin/bash

# 配信ストリーム名を指定
DELIVERY_STREAM_NAME="sample-s3-table-stream"

# リージョンを指定
AWS_REGION="us-east-1"

# ランダムなセンサーデータを生成する関数
generate_sensor_data() {
  local sensor_id=$1  # センサーIDを引数として受け取る
  # ランダムな温度 (20.00〜30.00)
  local temperature=$(echo "scale=2; 20 + (30 - 20) * $RANDOM / 32767" | bc)

  # ランダムな湿度 (30.00〜70.00)
  local humidity=$(echo "scale=2; 30 + (70 - 30) * $RANDOM / 32767" | bc)

  local timestamp=$(date -u +"%Y-%m-%dT%H:%M:%SZ")  # UTCタイムスタンプ

  # JSON形式でデータを生成
  jq -n --arg sensor_id "$sensor_id" \
        --arg temperature "$temperature" \
        --arg humidity "$humidity" \
        --arg timestamp "$timestamp" \
        '{sensor_id: $sensor_id, temperature: ($temperature | tonumber), humidity: ($humidity | tonumber), timestamp: $timestamp}'
}

# Base64エンコードを行う関数
base64_encode() {
  echo -n "$1" | base64
}

# データをFirehoseにバッチ送信する関数
send_batch_to_firehose() {
  local batch_data=$1
  local result

  # Firehoseにデータをバッチ送信し、結果をキャプチャ
  result=$(aws firehose put-record-batch \
    --region "$AWS_REGION" \
    --delivery-stream-name "$DELIVERY_STREAM_NAME" \
    --records "$batch_data" 2>&1)

  # 結果を確認
  if [ $? -eq 0 ]; then
    echo "Successfully sent batch: $batch_data"
    echo "AWS Response: $result"
  else
    echo "Failed to send batch: $batch_data"
    echo "Error: $result"
    exit 1
  fi
}

# 10秒ごとにセンサーごとのデータを10件まとめて送信
echo "Starting to send sensor data to Firehose..."
while true; do
  for sensor_id in {1..10}; do
    # センサーごとのデータを10件まとめる
    records=$(jq -n '[]')  # 空のJSON配列を初期化
    for i in {1..10}; do
      sensor_data=$(generate_sensor_data "$sensor_id")
      encoded_data=$(base64_encode "$sensor_data")
      # Firehose用のレコード形式に変換
      record=$(jq -n --arg data "$encoded_data" '{Data: $data}')
      records=$(echo "$records" | jq ". += [$record]")
    done

    # Firehose用のバッチデータをJSON文字列に変換
    batch_data=$(echo "$records" | jq -c '.')

    # Firehoseにバッチ送信
    send_batch_to_firehose "$batch_data"
  done

  sleep 10
done

chmod 744 stream-sensor-data.sh
./stream-sensor-data.sh &> output.log

spark.sql("select * from s3tablesbucket.sample.sensor_readings limit 5").show()

+---------+-----------+--------+-------------------+
|sensor_id|temperature|humidity|          timestamp|
+---------+-----------+--------+-------------------+
|        1|      28.77|   46.78|2024-12-06 21:00:56|
|        2|       22.2|   60.46|2024-12-06 21:00:59|
|        3|      26.73|   59.14|2024-12-06 21:01:01|
|        4|      25.55|   33.87|2024-12-06 21:01:04|
|        5|      22.44|   61.46|2024-12-06 21:01:06|
+---------+-----------+--------+-------------------+

select * from s3tablesbucket.sample.sensor_readings limit 5